/*
 * @Author: wuqingchun
 * @Date: 2023-08-19 17:57:14
 * @Description: 
 * @LastEditors: wuqingchun
 * @LastEditTime: 2024-01-11 15:44:48
 */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <sys/types.h>          /* See NOTES */
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <unistd.h>
 #include <sys/mman.h>

#include <sys/sendfile.h>


#include "ftpc.h"
#include "ftpc_log.h"
#include "ftpc_tcp.h"
#include "ftpc_event.h"
#include "ftpc_util.h"
#include "ftpc_dtp.h"

#define FTPC_DTP_BUF_MAXLEN 4096

#define FTPC_DTP_TPS_UNITS_KB 1024
#define FTPC_DTP_TPS_UNITS_MB 1024*1024

static inline unsigned long __dtp_GetTimeDiff(struct timeval* pstTValEnd, struct timeval* pstTValStart)
{
    struct timeval stTValTemp = *pstTValEnd;

    if (pstTValEnd->tv_usec < pstTValStart->tv_usec)
    {
        stTValTemp.tv_sec = pstTValEnd->tv_sec - 1;
        stTValTemp.tv_usec = pstTValEnd->tv_usec + 1000000;
    }

    stTValTemp.tv_sec -= pstTValStart->tv_sec;
    stTValTemp.tv_usec -= pstTValStart->tv_usec;
    return stTValTemp.tv_sec * 1000000 + stTValTemp.tv_usec;
}

static inline void __dtp_CloseListenSock(FTPC_DTP_CONTEXT_S* pstCtx)
{
    FTPC_LOG_INFO("Closed data listen connection, fd=%d", pstCtx->iListenSock);
    FTPC_event_DeleteFd(pstCtx->iListenSock);
    close(pstCtx->iListenSock);
    pstCtx->iListenSock= -1;
}

static inline void __dtp_CloseDataSock(FTPC_DTP_CONTEXT_S* pstCtx)
{
    FTPC_LOG_INFO("Closed data connection, fd=%d", pstCtx->iDataSock);
    FTPC_event_DeleteFd(pstCtx->iDataSock);
    close(pstCtx->iDataSock);
    pstCtx->iDataSock = -1;
}

static int __dtp_Accept(FTPC_DTP_CONTEXT_S* pstCtx)
{
    int iErr = -1;
    struct sockaddr_in stPeerAddr;
    socklen_t uiAddrLen = sizeof(stPeerAddr);

    pstCtx->iDataSock = accept(pstCtx->iListenSock, (struct sockaddr*)&stPeerAddr, &uiAddrLen);
    if (pstCtx->iDataSock == -1)
    {
        FTPC_LOG_ERROR("Failed to accept new client connection, %s", strerror(errno));
    }
    else
    {
        iErr = 0;
        FTPC_LOG_INFO("Succeed to accept data connection request for the client[%s:%hu]",
                      inet_ntoa(stPeerAddr.sin_addr), ntohs(stPeerAddr.sin_port));

        if (pstCtx->enTransferType == FTPC_TRANSFER_PUT)
        {
            FTPC_event_AddFd(pstCtx->iDataSock, FTPC_EVENT_OUT);
        }
        else
        {
            FTPC_event_AddFd(pstCtx->iDataSock, FTPC_EVENT_IN);
        }
    }

    __dtp_CloseListenSock(pstCtx);

    return iErr;
}

static int __dtp_Recv(FTPC_DTP_CONTEXT_S* pstCtx)
{
    int iErr;
    FTPC_BUF_HEAD_S* pstBufHead = &g_stFTPCClient.stDataBuf;

    FTPC_buf_Init(pstBufHead, FTPC_DTP_BUF_MAXLEN);
    
    // 接收数据
    iErr = FTPC_tcp_Recv(pstCtx->iDataSock, pstBufHead);
    if (iErr != 0)
    {
        FTPC_LOG_INFO("Failed to receive data, %s", strerror(errno));

        (void)gettimeofday(&pstCtx->stTValEnd, NULL);
        FTPC_dtp_CloseSock();
        FTPC_dtp_CloseTransferFd();
        FTPC_dtp_StopTransfer();
        
        return iErr;
    }

    // 写入文件或标准输出
    FTPC_BUF_S* pstBuf = pstBufHead->pstHead;
    while (pstBuf != NULL)
    {
        pstCtx->ulTransferLen += pstBuf->ulDataSize;
        if (pstCtx->iTransferFd != -1)
        {
            write(pstCtx->iTransferFd, pstBuf->pcBuf, pstBuf->ulDataSize);
        }
        
        //pstBuf->pcBuf[pstBuf->ulDataSize] = '\0';
        //printf("%s", pstBuf->pcBuf);

        pstBufHead->pstHead = pstBuf->pstNext;

        free(pstBuf);
        pstBuf = pstBufHead->pstHead;
    }

    pstBufHead->pstTail = NULL;

    return iErr;
}

static int __dtp_SendMMap(FTPC_DTP_CONTEXT_S* pstCtx)
{
    int iRet;
    unsigned long ulPerLen;
    unsigned long ulLeftLen = 0;
    ssize_t lSendResult;

    if (pstCtx->pAddr == NULL)
    {
        return -1;
    }

    iRet = 0;
    while (pstCtx->ulTransferLen < pstCtx->ulTotalLen)
    {
        ulPerLen = pstCtx->ulTotalLen - pstCtx->ulTransferLen;
        if (ulPerLen > 4096)
        {
            ulPerLen = 4096;
        }

        lSendResult = FTPC_tcp_Send(pstCtx->iDataSock, pstCtx->pAddr + pstCtx->ulTransferLen, ulPerLen);
        FTPC_LOG_DEBUG("SendResult:%ld, %d:%s", lSendResult, errno, strerror(errno));
        if (lSendResult == -1)
        {
            // 发送失败，需要关闭连接
            iRet = -1;
            break;
        }
        
        pstCtx->ulTransferLen += lSendResult;

        if (lSendResult != ulPerLen)
        {
            // 发送缓冲区满，暂停发送，等待异步事件通知继续发送
            FTPC_event_AddFd(pstCtx->iDataSock, FTPC_EVENT_OUT);
            iRet = 1;
            break;
        }

        // 数据全部发送成功，读取数据继续发送
    }

    if (iRet <= 0)
    {
        //printf("secs: %lu, usecs: %lu");
        (void)gettimeofday(&pstCtx->stTValEnd, NULL);

        // 全部发送成功、发送失败都需要关闭连接
        __dtp_CloseDataSock(pstCtx);
    }   
        
    return 0;
}

static int __dtp_Sendfile(FTPC_DTP_CONTEXT_S* pstCtx)
{
    int iRet;
    off_t lOffset;
    size_t ulCount = 0x7ffff000;
    ssize_t lSendSize;

    while (pstCtx->ulTransferLen < pstCtx->ulTotalLen)
    {
        ulCount = pstCtx->ulTotalLen - pstCtx->ulTransferLen;
        if (ulCount > FTPC_DTP_TPS_UNITS_MB)
        {
            ulCount = FTPC_DTP_TPS_UNITS_MB;
        }

        lOffset = pstCtx->ulTransferLen;
        lSendSize = sendfile(pstCtx->iDataSock, pstCtx->iTransferFd, &lOffset, ulCount);
        if (lSendSize == -1)
        {
            // 发送失败，需要关闭连接
            iRet = -1;
            break;
        }

        pstCtx->ulTransferLen += lSendSize;

        if (lSendSize != ulCount)
        {
            // 发送缓冲区满，暂停发送，等待异步事件通知继续发送
            FTPC_event_AddFd(pstCtx->iDataSock, FTPC_EVENT_OUT);
            iRet = 1;
            break;
        }

        // 数据全部发送成功，读取数据继续发送
    }

    if (iRet <= 0)
    {
        (void)gettimeofday(&pstCtx->stTValEnd, NULL);

        // 全部发送成功、发送失败都需要关闭连接
        __dtp_CloseDataSock(pstCtx);
    }    

    return 0;
}

static int __dtp_Send(FTPC_DTP_CONTEXT_S* pstCtx)
{
    int iRet;
    ssize_t lReadSize;
    char acBuf[4096];
    ssize_t lSendResult;

    iRet = 0;
    while (pstCtx->ulTransferLen < pstCtx->ulTotalLen)
    {
        lReadSize = read(pstCtx->iTransferFd, acBuf, 4096);
        if (lReadSize > 0)
        {
            lSendResult = FTPC_tcp_Send(pstCtx->iDataSock, acBuf, lReadSize);
            FTPC_LOG_DEBUG("SendResult:%ld, %d:%s", lSendResult, errno, strerror(errno));
            if (lSendResult == -1)
            {
                // 发送失败，需要关闭连接
                iRet = -1;
                break;
            }
            
            pstCtx->ulTransferLen += lSendResult;

            if (lSendResult != lReadSize)
            {
                // 发送缓冲区满，暂停发送，等待异步事件通知继续发送
                FTPC_event_AddFd(pstCtx->iDataSock, FTPC_EVENT_OUT);
                iRet = 1;
                break;
            }

            // 数据全部发送成功，读取数据继续发送
        }
        else
        {
            FTPC_LOG_ERROR("Failed to read data from file, fd=%d, %s", pstCtx->iTransferFd, strerror(errno));
            iRet = -1;
            break;
        }
    }

    if (iRet <= 0)
    {
        (void)gettimeofday(&pstCtx->stTValEnd, NULL);

        // 全部发送成功、发送失败都需要关闭连接
        __dtp_CloseDataSock(pstCtx);
    }    

    return 0;
}

static int __dtp_SendDirectIO(FTPC_DTP_CONTEXT_S* pstCtx)
{
    int iRet;
    ssize_t lReadSize;
    char* pcBuf = pstCtx->pcDirectIO;
    ssize_t lSendResult;

    iRet = 0;
    while (pstCtx->ulTransferLen < pstCtx->ulTotalLen)
    {
        lReadSize = read(pstCtx->iTransferFd, pcBuf, 4096);
        if (lReadSize > 0)
        {
            lSendResult = FTPC_tcp_Send(pstCtx->iDataSock, pcBuf, lReadSize);
            FTPC_LOG_DEBUG("SendResult:%ld, %d:%s", lSendResult, errno, strerror(errno));
            if (lSendResult == -1)
            {
                // 发送失败，需要关闭连接
                iRet = -1;
                break;
            }
            
            pstCtx->ulTransferLen += lSendResult;

            if (lSendResult != lReadSize)
            {
                // 发送缓冲区满，暂停发送，等待异步事件通知继续发送
                FTPC_event_AddFd(pstCtx->iDataSock, FTPC_EVENT_OUT);
                iRet = 1;
                break;
            }

            // 数据全部发送成功，读取数据继续发送
        }
        else
        {
            FTPC_LOG_ERROR("Failed to read data from file, fd=%d, %s", pstCtx->iTransferFd, strerror(errno));
            iRet = -1;
            break;
        }
    }

    if (iRet <= 0)
    {
        (void)gettimeofday(&pstCtx->stTValEnd, NULL);

        // 全部发送成功、发送失败都需要关闭连接
        __dtp_CloseDataSock(pstCtx);
    }    

    return 0;
}

int FTPC_dtp_Input(int iSockFd)
{
    int iErr;
    FTPC_DTP_CONTEXT_S* pstCtx = &g_stFTPCClient.stDTPCtx;
    
    if (iSockFd == pstCtx->iListenSock)
    {
        FTPC_LOG_DEBUG("Input event in data listen connection, fd=%d", iSockFd);
        iErr = __dtp_Accept(pstCtx);
    }
    else
    {
        FTPC_LOG_DEBUG("Input event in data connection, fd=%d", iSockFd);
        iErr = __dtp_Recv(pstCtx);
    }

    return iErr;
}

int FTPC_dtp_Output(int iSockFd)
{
    int iErr = -1;
    FTPC_DTP_CONTEXT_S* pstCtx = &g_stFTPCClient.stDTPCtx;
    
    FTPC_LOG_DEBUG("Output event in data connection, fd=%d", iSockFd);

    if (pstCtx->iTransferFd == -1 || iSockFd != pstCtx->iDataSock)
    {
        return iErr;
    }

    switch (pstCtx->enSendMode)
    {
        case FTPC_SEND_MODE_SENDFILE:
        {
            iErr = __dtp_Sendfile(pstCtx);
            break;
        }
        case FTPC_SEND_MODE_MMAP:
        {
            iErr = __dtp_SendMMap(pstCtx);
            break;
        }
        case FTPC_SEND_MODE_DIRECT_IO:
        {
            iErr = __dtp_SendDirectIO(pstCtx);
            break;
        }
        case FTPC_SEND_MODE_NORMAL:
        default:
        {
            iErr = __dtp_Send(pstCtx);
            break;
        }
    }

    return iErr;
}

void FTPC_dtp_CloseSock(void)
{
    FTPC_DTP_CONTEXT_S* pstCtx = &g_stFTPCClient.stDTPCtx;
    
    if (pstCtx->iListenSock != -1)
    {
        __dtp_CloseListenSock(pstCtx);
    }

    if (pstCtx->iDataSock != -1)
    {
        __dtp_CloseDataSock(pstCtx);
    }
}

void FTPC_dtp_CloseTransferFd(void)
{
    FTPC_DTP_CONTEXT_S* pstCtx = &g_stFTPCClient.stDTPCtx;
    
    if (pstCtx->iTransferFd != -1 && pstCtx->iTransferFd != STDOUT_FILENO)
    {
        close(pstCtx->iTransferFd);
        pstCtx->iTransferFd = -1;
    }

    if (pstCtx->pAddr != NULL)
    {
        (void)munmap(pstCtx->pAddr, pstCtx->ulTotalLen);
        pstCtx->pAddr = NULL;
    }

    if (pstCtx->pcDirectIO != NULL)
    {
        free(pstCtx->pcDirectIO);
        pstCtx->pcDirectIO = NULL;
    }
}

void FTPC_dtp_StartTransfer(int iTransferFd)
{
    struct stat stStat;
    FTPC_DTP_CONTEXT_S* pstCtx = &g_stFTPCClient.stDTPCtx;

    if (pstCtx->iTransferFd != -1 && pstCtx->iTransferFd != STDOUT_FILENO)
    {
        close(pstCtx->iTransferFd);
        pstCtx->iTransferFd = -1;
    }

    pstCtx->iTransferFd = iTransferFd;
    pstCtx->ulTransferLen = 0;
    pstCtx->ulTotalLen = 0;

    pstCtx->pAddr = NULL;

    if (pstCtx->enTransferType == FTPC_TRANSFER_PUT)
    {
        FTPC_event_AddFd(pstCtx->iDataSock, FTPC_EVENT_OUT);
        if (fstat(iTransferFd, &stStat) == 0)
        {
            pstCtx->ulTotalLen = stStat.st_size;

            if (pstCtx->enSendMode == FTPC_SEND_MODE_MMAP)
            {
                pstCtx->pAddr = mmap(NULL, pstCtx->ulTotalLen, PROT_READ, MAP_PRIVATE, pstCtx->iTransferFd, 0);
                if (pstCtx->pAddr == MAP_FAILED)
                {
                    FTPC_LOG_ERROR("mmap error, %s", strerror(errno));
                }
            }
            else if (pstCtx->enSendMode == FTPC_SEND_MODE_DIRECT_IO)
            {
                int iRet = posix_memalign((void**)&pstCtx->pcDirectIO, 512, 4096);
                if (iRet == 0)
                {
                    FTPC_util_EnableDirectIO(iTransferFd);
                    FTPC_LOG_INFO("Enable direct io mode, %s", strerror(errno));
                }
                else
                {
                    FTPC_LOG_ERROR("posix_memalign error, %s", strerror(errno));
                    __dtp_CloseDataSock(pstCtx);
                }                
            }
        }
    }
    else
    {
        FTPC_event_AddFd(pstCtx->iDataSock, FTPC_EVENT_IN);
    }
    
    pstCtx->stTValEnd.tv_sec = 0;
    pstCtx->stTValEnd.tv_usec = 0;
    
    (void)gettimeofday(&pstCtx->stTValStart, NULL);
}

void FTPC_dtp_StopTransfer(void)
{
    FTPC_DTP_CONTEXT_S* pstCtx = &g_stFTPCClient.stDTPCtx;
    double dfTimeSec;
    double dTps;
    char* pcUnits = "B/s";

    if (pstCtx->enTransferType == FTPC_TRANSFER_LIST)
    {
        return;
    }
    
    if (pstCtx->ulTransferLen > 0 && pstCtx->iDataSock == -1)
    {
        dfTimeSec = __dtp_GetTimeDiff(&pstCtx->stTValEnd, &pstCtx->stTValStart)/1000000.0;
        dTps = pstCtx->ulTransferLen / dfTimeSec;

        if (dTps >= FTPC_DTP_TPS_UNITS_MB)
        {
            pcUnits = "MB/s";
            dTps /= FTPC_DTP_TPS_UNITS_MB;
        }
        else if (dTps >= FTPC_DTP_TPS_UNITS_KB)
        {
            pcUnits = "KB/s";
            dTps /= FTPC_DTP_TPS_UNITS_KB;
        }

        if (pstCtx->ulTotalLen == 0)
        {
            printf("%lu bytes received in %0.2lf secs (%lf %s)\n", pstCtx->ulTransferLen, 
                    dfTimeSec, dTps, pcUnits);
        }
        else
        {
            printf("%lu bytes sent in %0.2lf secs (%lf %s)\n", pstCtx->ulTransferLen, 
                   dfTimeSec, dTps, pcUnits);
        }

        pstCtx->ulTotalLen = 0;
        pstCtx->ulTransferLen = 0;
    }
}

int FTPC_dtp_Listen(int iCtlSock, struct sockaddr_in* pstAddr)
{
    int iRet;
    socklen_t uiLAddrLen;
    FTPC_DTP_CONTEXT_S* pstCtx = &g_stFTPCClient.stDTPCtx;
    
    FTPC_dtp_CloseSock();

    uiLAddrLen = sizeof(struct sockaddr_in);
    iRet = getsockname(iCtlSock, (struct sockaddr*)pstAddr, &uiLAddrLen);
    if (iRet != 0)
    {
        FTPC_LOG_ERROR("getsockname error, %s", strerror(errno));
        return -1;
    }

    pstAddr->sin_port = 0; // 由操作系统选择一个可用的端口号进行绑定
    pstCtx->iListenSock = FTPC_tcp_Listen(pstAddr);
    if (pstCtx->iListenSock == -1)
    {
        FTPC_LOG_ERROR("Failed to listen, %s\n", strerror(errno));
        return -1;
    }
    
    FTPC_event_AddFd(pstCtx->iListenSock, FTPC_EVENT_IN);

    return 0;
}

int FTPC_dtp_Connect(const char *pcHostName, unsigned short usPort)
{
    FTPC_DTP_CONTEXT_S* pstCtx = &g_stFTPCClient.stDTPCtx;
    
    pstCtx->iDataSock = FTPC_tcp_Connect(pcHostName, usPort);
    if (pstCtx->iDataSock == -1)
    {
        return -1;
    }

    return 0;
}

void FTPC_dtp_SetTransferType(FTPC_TRANSFER_TYPE_E enType)
{
    g_stFTPCClient.stDTPCtx.enTransferType = enType;
}

FTPC_TRANSFER_TYPE_E FTPC_dtp_GetTransferType(void)
{
    return g_stFTPCClient.stDTPCtx.enTransferType;
}

void FTPC_dtp_Init(void)
{
    FTPC_DTP_CONTEXT_S* pstCtx = &g_stFTPCClient.stDTPCtx;
    
    pstCtx->enTransferType = FTPC_TRANSFER_MAX;
    pstCtx->enSendMode = FTPC_SEND_MODE_NORMAL;
    pstCtx->iListenSock = -1;
    pstCtx->iDataSock = -1;
    pstCtx->ulTotalLen = 0;
    pstCtx->ulTransferLen = 0;
    pstCtx->iTransferFd = -1;
    pstCtx->pAddr = NULL;
    pstCtx->pcDirectIO = NULL;

    FTPC_buf_Init(&g_stFTPCClient.stDataBuf, FTPC_DTP_BUF_MAXLEN);
}